package com.gxljc.bear.crawler.itaogao.cbcomcn;

import com.mongodb.BasicDBObject;
import com.mongodb.Bytes;
import com.mongodb.DBCursor;
import com.mongodb.DBObject;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.commons.util.Pair;
import com.gxljc.bear.crawler.base.*;
import com.gxljc.bear.crawler.itaogao.ItaogaoConst;
import com.gxljc.bear.crawler.itaogao.NewsPageTable;
import com.gxljc.bear.crawler.itaogao.jingji21.Jingji21PageCrawler;
import com.gxljc.bear.crawler.itaogao.util.HbaseUtil;
import com.gxljc.bear.crawler.itaogao.util.MongodbUtil;
import com.gxljc.bear.crawler.util.DateUtil;
import com.gxljc.bear.crawler.util.HtmlUtil;
import com.gxljc.bear.crawler.util.SparkUtil;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import java.io.Serializable;
import java.net.URLEncoder;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

/**
 * 南方日报 PAGE 数据爬虫。
 *
 * @author tanghaitao
 * @since 2022-9-6
 */
public class CbcomcnPageCrawler implements Serializable {
    public static Logger LOG = Logger.getLogger(Jingji21PageCrawler.class);
    private Boolean proxy = false;
    private final int CONTENT_LESS_NUM = 80; //内容少于多少即放弃
    private Map<String, String> proper = new HashMap<String, String>();

    public CbcomcnPageCrawler(Boolean proxy) {
//        proper.put("Cookie", CookieAccount.getInstance().getCbcomcnCookie());
//        proper.put("Cookie", "__jsluid=2319357cf501b43b1a787466f6fad34e; __jsl_clearance=1474556645.531|0|DkHCWQLDmbBYj01fp9cbywAcONc%3D");
        proper.put("Host", "dianzibao.cb.com.cn");
        this.proxy = proxy;
    }

    public void crawl() throws Exception {
        List<NewsPageTable> seeds = getSeed();
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        crawlNoSpark(seeds);
    }

    public void crawlNoSpark(List<NewsPageTable> seeds) {
        if (CollectionUtils.isEmpty(seeds)) {
            LOG.error("page is empty");
            return;
        }
        for (NewsPageTable table : seeds) {
            try {
                crawl(table);
            } catch (Exception e) {
                e.printStackTrace();  //To change body of catch statement use File | Settings | File Templates.
            }
        }
    }

    public void crawl(List<NewsPageTable> seeds) {
        int coreMax = 10;
        JavaSparkContext jsc = SparkUtil.createCommonsSparkContext(
                "bear-中国经营报 page详情爬取-" + seeds.size(), coreMax,
                coreMax * 2, CbcomcnPageCrawler.class);
        JavaRDD<NewsPageTable> seedsRDD = jsc.parallelize(new ArrayList<NewsPageTable>(seeds));
        long count = seedsRDD.mapPartitions(
                new FlatMapFunction<Iterator<NewsPageTable>, Integer>() {
                    private static final long serialVersionUID = 1L;

                    @Override
                    public Iterable<Integer> call(Iterator<NewsPageTable> seeds)
                            throws Exception {
                        int successCnt = 0;
                        while (seeds.hasNext()) {
                            try {
                                NewsPageTable seed = seeds.next();
                                int ret = crawl(seed);
                                if (ret == 1)
                                    successCnt++;
                                if (successCnt % 100 == 0) {
                                    LOG.info("successCnt=" + successCnt);
                                }
                            } catch (Exception er) {
                                er.printStackTrace();
                            }
                        }
                        return Arrays.asList(successCnt);
                    }
                }).count();
        LOG.info("all count=" + count);
        jsc.stop();
    }

    //单条url入口
    public int crawl(NewsPageTable table) throws Exception {
        LOG.info("crawl = " + table.crawlUrl);
        String url = table.getCrawlUrl();
        String newUrl = genCrawlUrl(url);
        DolphinFetchData fetchData = DolphinFetchData.getInstance(DolphinCrawlerConsts.CrawlerChannelType.bear.getName());
        byte[] htmlByte = fetchData.getHtml(newUrl, proxy, ItaogaoConst.CRAWL_TIMEOUT, proper);
        if (htmlByte == null) {
            LOG.error("page result is empty url = " + url);
            return DolphinCrawlerConsts.RESULT_NO;
        }
        String html = new String(htmlByte, "utf-8");
        extract(table, html);
        HbaseUtil.saveHbase(table, htmlByte);
        MongodbUtil.saveMongodb(table, ItaogaoConst.MONGODB_TABLE.CBCOMCN.getValue());
        return DolphinCrawlerConsts.RESULT_YES;
    }

    private String genCrawlUrl(String url) throws Exception {
        Configuration conf = DolphinCrawlerConf.getInstance();
        String webHost = conf.get(DolphinCrawlerConsts.PARAM_DETECTOR_WEB_HOST);
        if (StringUtils.isEmpty(webHost)) return url;
        String newUrl = URLEncoder.encode(url, "utf-8");
        return String.format(webHost + ItaogaoConst.DETECTOR_WEB_PATH, DolphinCrawlerConsts.PARAM_DETECTOR_WEB_ID, DolphinCrawlerConsts.PARAM_DETECTOR_WEB_TYPE, newUrl);
    }

    //内容抽取
    private void extractContent(NewsPageTable table, Document doc) {
        Elements elements = doc.select("founder-content p");
        if (CollectionUtils.isEmpty(elements)) return;
        StringBuffer sb = new StringBuffer();
        for (Element element : elements) {
            String content = element.text();
            if (sb.length() > 0) sb.append(DolphinCrawlerConsts.DEFAULT_LINE_BREAK_WORD);
            content = content.trim();
            while (content.startsWith(" ")) {
                content = content.substring(1);
            }
            sb.append(content);
        }
        table.setContent(sb.toString());
    }

    //发布时间抽取
    private void extractPublishDate(NewsPageTable table) {
        String url = table.crawlUrl;
        final String parse = "/html/(\\d+-\\d+/\\d+)/content_";
        Pattern pattern = Pattern.compile(parse);
        Matcher m = pattern.matcher(url);
        if (m.find()) {
            String word = m.group(1);
            Date date = DateUtil.parseFormat(word, "yyyy-MM/dd");
            table.setPublishDate(date);
        }
    }

    //标题抽取
    private void extractTitle(NewsPageTable table, Document doc) {
        Elements titleElements = doc.select(".font01");
        if (CollectionUtils.isEmpty(titleElements)) return;
        String title = titleElements.text();
        if (StringUtils.isEmpty(title)) return;
        table.setTitle(title);

        Elements subTitleElements = doc.select(".font02");
        if (CollectionUtils.isEmpty(subTitleElements)) return;
        for (Element subTitle : subTitleElements) {
            String sub = subTitle.text();
            if (!StringUtils.isEmpty(sub)) {
                table.setSubTitle(sub);
            }
        }
    }


    //作者抽取
    private void extractAuthor(NewsPageTable table) {
        try {
            String author = null;
            if (StringUtils.isEmpty(table.content)) return;
            if (table.content.length() < CONTENT_LESS_NUM) return;
            if (table.content.indexOf(DolphinCrawlerConsts.DEFAULT_LINE_BREAK_WORD) < 0) return;
            String[] lines = table.content.split(DolphinCrawlerConsts.DEFAULT_LINE_BREAK_WORD);
            for (String line : lines) {
                author = parseAuthor(line);
                if (!StringUtils.isEmpty(author)) {
                    table.setAuthor(author);
                    return;
                }
            }
        } catch (Exception er) {
            er.printStackTrace();
            LOG.error("extract author error, url=" + table.crawlUrl, er);
        }
    }

    //抽取作者
    private String parseAuthor(String text) {
        final String[] parse = {"本报记者[ ]+(.*?)[ ]+"};
        for (String p : parse) {
            Pattern pattern = Pattern.compile(p);
            Matcher m = pattern.matcher(text);
            if (m.find()) {
                return m.group(1);
            }
        }
        return null;
    }

    //缩略图抽取
    private void extractImages(NewsPageTable table, Document doc) throws Exception {
        Elements elements = doc.select(".font6 table img");
        if (CollectionUtils.isEmpty(elements)) return;
        List<String> images = new LinkedList<>();
        for (Element element : elements) {
            String src = element.attr("src");
            if (!StringUtils.isEmpty(src)) {
                String newUrl = HtmlUtil.pictureUrlPref(src, table.crawlUrl);
//                ImageFresh fresh = ImageFresh.getInstance();
//                Pair<Integer, String> newPicPair = fresh.freshOne(newUrl, proper);
//                String newPic = newPicPair.second;
                String newPic = newUrl;
                if (StringUtils.isEmpty(table.logoUrl))
                    table.setLogoUrl(newPic);
                images.add(newPic);
            }
        }
        table.setImages(images);
    }

    //抽取object
    private void extract(NewsPageTable table, String html) throws Exception {
        Document doc = Jsoup.parse(html);
        extractTitle(table, doc);
        extractContent(table, doc);
        extractPublishDate(table);
        extractImages(table, doc);
        extractAuthor(table);
        table.setStatus(0);
        table.setCrawlFlag(1);
    }

    //获取seed
    public List<NewsPageTable> getSeed() throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.ITAOGAO.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(DolphinCrawlerConsts.MONGODB_DEFAULT_ID, 1);
        keys.put(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue(), 1);
        DBObject query = new BasicDBObject();
        query.put(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_FLAG.getValue(), new BasicDBObject("$ne", 1));
        DBCursor cursor = repo.getCollection(ItaogaoConst.MONGODB_TABLE.CBCOMCN.getValue())
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<NewsPageTable> seeds = new LinkedList<>();
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object id = data.get(DolphinCrawlerConsts.MONGODB_DEFAULT_ID);
                Object url = data.get(ItaogaoConst.SCHEMA_COLUMN_NAME.CRAWL_URL.getValue());
                NewsPageTable table = new NewsPageTable();
                table.setId(id.toString());
                table.setCrawlUrl(url.toString());
                seeds.add(table);
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return seeds;
    }
}
